package org.tao.lightningmq.nameserver.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.dto.HeartBeatDTO;
import org.tao.lightningmq.common.dto.ServiceRegistryRespDTO;
import org.tao.lightningmq.common.dto.SlaveAckDTO;
import org.tao.lightningmq.common.enums.NameServerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.nameserver.common.CommonCache;
import org.tao.lightningmq.nameserver.enums.ReplicationMsgTypeEnum;
import org.tao.lightningmq.nameserver.event.model.SlaveReplicationMsgAckEvent;

/**
 * @Author lht
 * @date 2025/1/16 - 22:03
 * @description: 主节点接受从节点ACK
 */
public class ReplicationSlaveMsgAckListener implements Listener<SlaveReplicationMsgAckEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ReplicationSlaveMsgAckListener.class);

    @Override
    public void onReceive(SlaveReplicationMsgAckEvent event) throws Exception {
        String msgId = event.getMsgId();
        SlaveAckDTO slaveAckDTO = CommonCache.getAckMap().get(msgId);
        if (slaveAckDTO == null) {
            return;
        }
        int curAckTime = slaveAckDTO.getNeedAckNum().decrementAndGet();
        if (curAckTime == 0) {
            LOGGER.info("主节点接收所有从节点ACK成功，msgId:{}", msgId);
            CommonCache.getAckMap().remove(msgId);
            if (event.getType() == ReplicationMsgTypeEnum.REGISTRY.getCode()) {
                ServiceRegistryRespDTO respDTO = new ServiceRegistryRespDTO();
                respDTO.setMsgId(msgId);
                slaveAckDTO.getBrokerChannel().writeAndFlush(new TcpMsg(
                        NameServerResponseCode.REGISTRY_SUCCESS.getCode(),
                        JSON.toJSONBytes(respDTO)
                ));
            } else if (event.getType() == ReplicationMsgTypeEnum.HEART_BEAT.getCode()) {
                HeartBeatDTO heartBeatDTO = new HeartBeatDTO();
                heartBeatDTO.setMsgId(msgId);
                slaveAckDTO.getBrokerChannel().writeAndFlush(new TcpMsg(
                        NameServerResponseCode.HEART_BEAT_SUCCESS.getCode(),
                        JSON.toJSONBytes(heartBeatDTO)
                ));
            }

        }
    }
}
